package dataflow.core;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.log4j.Logger;

public class Storage extends PE {

	static protected Logger log = Logger.getLogger(Storage.class);
	
	protected Queue<Integer> iterationQueue = new LinkedList<Integer>();
	
	private AtomicBoolean sourceIsRunning = new AtomicBoolean(false);
	
	public Storage(dataflowScheme.Storage estorage, LinkedBlockingQueue<IEvent> queue, LinkedBlockingQueue<ExecutionEvent> executionQueue){
		super(estorage, queue, executionQueue);
	}

	protected void run(int portset) {
		ExecutionEvent event = new StorageExecutionEvent(this, portset);
		executionQueue.offer(event);
	}

	@Override
	protected Integer getNextPortset2invokeRun() {
		return iterationQueue.poll();
	}

	@Override
	protected void invokeNextIteration(int portSet) {
		if(sourceIsRunning.get()){
			//plan iteration
			iterationQueue.add(portSet);
		} else {
			run(portSet);
		}		
	}

	@Override
	protected void endOfSourceRun() {
		sourceIsRunning.set(false);
	}

}
